import { Application } from "egg";
import CodoBase from "./CodoBase";
import { DateFormat, sleep } from "@diyaner/ding";
import CodoError from "./CodoError";

const stopedSymbol = Symbol("#CodoMQ.stoped");
const isRunningSymbol = Symbol("#CodoMQ.running");

type Message = {
	error?: string;
	data: Record<string, any>;
	/** 上次消费时间，如果失败重试，就能知道 */
	time?: string;
	retryTimes: number; // 重试次数
};

/** 一个基于redis的mini MQ实现，简化事件触发、订阅处理、错误重试等功能 */
export default class CodoMQ extends CodoBase {
	constructor(app: Application, public readonly namespace: string) {
		super(app);
	}

	/** 此属性用于外部带出消息格式对象，不做其他用 */
	messageType: Message;

	get redisKey() {
		return `codo:mqlist:${this.namespace}`;
	}

	/** 队列锁的key */
	get redisKeyLock() {
		return `${this.redisKey}:locked`;
	}
	/** 消费失败的消息存放 set 集合 */
	get redisKeyFailed() {
		return `${this.redisKey}:failed`;
	}

	/** 向队列尾添加消息 */
	async push(...msgs: Record<string, any>[]) {
		const msgs2 = msgs.map((value) => {
			const obj: Message = {
				data: value,
				retryTimes: 0,
			};
			return JSON.stringify(obj);
		});
		return await this.app.redis.rpush(this.redisKey, ...msgs2);
	}

	private [stopedSymbol] = false;
	stop() {
		this[stopedSymbol] = true;
		this[isRunningSymbol] = false;
	}
	/** 停止后，又继续消费 */
	resume() {
		this[stopedSymbol] = false;
		this.start(); // 继续消费
	}

	public consumerCallback: <T extends Record<string, any>>(msg: T) => Promise<void>;
	/** 为本实例设置一个消费者回调函数，用于执行处理消息 */
	setConsumer(cb: (msg: any) => Promise<void>) {
		this.consumerCallback = cb;
	}

	/** 无消息时等待毫秒数 */
	noMessageWaitMilliSeconds = 1500;

	/** 是否已开始消费，开始消费不能重复消费，停止后重新消费可以继续。 */
	private [isRunningSymbol] = false;

	/** 开始消费，请先设置消费终止条件，否则将永远循环消费 */
	async start() {
		if (this[isRunningSymbol]) throw new CodoError("consumer is running");
		this[isRunningSymbol] = true;

		if (!this.consumerCallback) {
			throw new CodoError("consumerCallback is not set");
		}

		/** 其他异常错误捕获 */
		const onFailedHandler = async (error: Error) => {
			try {
				this.app.logger.error(`[CodoMQ] 队列循环消费出错 error: %s`, error);
				await this.app.redis.del(this.redisKeyLock);
			} catch (error) {
				this.app.logger.error(`[CodoMQ] 队列循环消费出错后，取消redisLock可能出错 error: %s`, error);
			}
			await sleep(1500);
		};
		while (!this[stopedSymbol]) {
			try {
				if (await this.app.redis.exists(this.redisKeyLock)) {
					await sleep(660);
					continue;
				}
				await this.app.redis.set(this.redisKeyLock, DateFormat.init().format(), "EX", 60 * 10); // 处理一个消息时，整个list加锁，防止其他消费者重复消费
				const msg = await this.app.redis.lindex(this.redisKey, 0);
				if (!msg) {
					await this.app.redis.del(this.redisKeyLock);
					await sleep(this.noMessageWaitMilliSeconds || 1500);
					continue;
				}

				let objMsg: Message;

				try {
					objMsg = JSON.parse(msg);
				} catch (error) {
					this.app.logger.error(
						`[CodoMQ] 队列消费一个消息，消息格式错误，不是json对象，消息将被丢弃 \r\n%s\r\n%s`,
						msg,
						error
					);
					await this.app.redis.lpop(this.redisKey);
					await this.app.redis.del(this.redisKeyLock);
					await sleep(1000);
					continue;
				}

				try {
					await this.consumerCallback(objMsg.data);
				} catch (err: any) {
					this.app.logger.error(`[CodoMQ] 队列消费一个消息，消费者出错 error: %s`, err);
					const msg2: Message = {
						data: objMsg.data,
						retryTimes: objMsg.retryTimes,
						error: err?.message || "unknown",
						time: DateFormat.init().format(),
					};
					await this.app.redis.sadd(this.redisKeyFailed, JSON.stringify(msg2));
				}

				await this.app.redis.lpop(this.redisKey);
				await this.app.redis.del(this.redisKeyLock);
			} catch (error: any) {
				await onFailedHandler(error);
			}
		}
		this[isRunningSymbol] = false;
	}

	/**
	 * 消费失败队列重新消费
	 *
	 * @param {number} times 限定仅小于此次数的失败消息重新消费
	 * @param {Date} earlyTime 限定仅小于该时间的失败消息重新消费，如果同时提供次数和时间，则两个条件都满足才重消费
	 * @memberof CodoMQ
	 */
	async reConsumerFailed(times: number, earlyTime?: Date) {
		let index: string = "0";
		let member: [string, string[]] | undefined;

		const waitRemoveMember: string[] = [];

		const handleMsg = async (msg1: string) => {
			const msg: Message = JSON.parse(msg1);
			if (msg.retryTimes > times) return;
			if (earlyTime) {
				if (DateFormat.init(msg.time).getTime() > DateFormat.init(earlyTime).getTime()) return;
			}
			msg.retryTimes++;
			await this.app.redis.rpush(this.redisKey, JSON.stringify(msg));
			waitRemoveMember.push(msg1);
		};
		while (true) {
			member = await this.app.redis.sscan(this.redisKeyFailed, index, "COUNT", 5);
			if (!member) break;
			index = member[0];
			for (const msg of member[1]) {
				try {
					await handleMsg(msg);
				} catch (error) {
					this.app.logger.error(`[CodoMQ] 失败消息重新加入消费队列，出错 error:\r\n%s\r\n%s`, msg, error);
				}
			}
			if (member && member[0] == "0") break;
		}
		if (waitRemoveMember.length > 0) {
			await this.app.redis.srem(this.redisKeyFailed, ...waitRemoveMember);
		}
	}

	/**
	 * 消费失败队列重新消费，通过时间控制，不通过失败次数
	 *
	 * @param {Date} earlyTime 限定仅小于该时间的失败消息重新消费
	 * @memberof CodoMQ
	 */
	async reConsumerFailedByDate(earlyTime: Date) {
		let index: string = "0";
		let member: [string, string[]] | undefined;

		const waitRemoveMember: string[] = [];

		const handleMsg = async (msg1: string) => {
			const msg: Message = JSON.parse(msg1);
			if (earlyTime) {
				if (DateFormat.init(msg.time).getTime() > DateFormat.init(earlyTime).getTime()) return;
			}
			msg.retryTimes++;
			await this.app.redis.rpush(this.redisKey, JSON.stringify(msg));
			waitRemoveMember.push(msg1);
		};
		while (true) {
			member = await this.app.redis.sscan(this.redisKeyFailed, index, "COUNT", 5);
			if (!member) break;
			index = member[0];
			for (const msg of member[1]) {
				try {
					await handleMsg(msg);
				} catch (error) {
					this.app.logger.error(`[CodoMQ] 失败消息重新加入消费队列，出错 error:\r\n%s\r\n%s`, msg, error);
				}
			}
			if (member && member[0] == "0") break;
		}
		if (waitRemoveMember.length > 0) {
			await this.app.redis.srem(this.redisKeyFailed, ...waitRemoveMember);
		}
	}

	/** 清除大于等于该重试次数或小于某执行时间的消费错误的消息 */
	async clearFailed(timesOrEarlyTime: number | Date) {
		let index: string = "0";
		let member: [string, string[]] | undefined;

		const times = typeof timesOrEarlyTime === "number" ? timesOrEarlyTime : 0;
		const earlyTime = typeof timesOrEarlyTime === "number" ? undefined : timesOrEarlyTime;
		let count = 0;
		const handleMsg = async (msg1: string) => {
			const msg: Message = JSON.parse(msg1);
			if (times && msg.retryTimes >= times) {
				await this.app.redis.srem(this.redisKeyFailed, msg1);
				count++;
			}
			if (earlyTime) {
				if (DateFormat.init(msg.time).getTime() < DateFormat.init(earlyTime).getTime()) {
					await this.app.redis.srem(this.redisKeyFailed, msg1);
					count++;
				}
			}
		};
		while (true) {
			member = await this.app.redis.sscan(this.redisKeyFailed, index, "COUNT", 5);
			if (!member) break;
			index = member[0];
			for (const msg of member[1]) {
				try {
					await handleMsg(msg);
				} catch (error) {
					this.app.logger.error(`[CodoMQ] 失败消息移除，出错 error:\r\n%s\r\n%s`, msg, error);
				}
			}
			if (member && member[0] == "0") break;
		}
		return count;
	}

	/** 获取某个重复消费次数的集合 */
	async getFailedList(times: number = 0) {
		let index: string = "0";
		let member: [string, string[]] | undefined;

		const list: Message[] = [];
		const handleMsg = async (msg1: string) => {
			const msg: Message = JSON.parse(msg1);
			if (msg.retryTimes == times) {
				list.push(msg);
			}
		};
		while (true) {
			member = await this.app.redis.sscan(this.redisKeyFailed, index, "COUNT", 5);
			console.log(member);
			if (!member) break;
			index = member[0];
			for (const msg of member[1]) {
				try {
					await handleMsg(msg);
				} catch (error) {
					this.app.logger.error(`[CodoMQ] 获取某个消费失败次数，出错 error:\r\n%s\r\n%s`, msg, error);
				}
			}
			if (member && member[0] == "0") break;
		}
		return list;
	}

	/** 获取队列数量 */
	async getCount() {
		const key = this.redisKey;
		return await this.app.redis.llen(key);
	}
}
